package git.soulbgm.controller;

import cn.hutool.core.util.StrUtil;
import git.soulbgm.common.base.BaseController;
import git.soulbgm.monitor.KafkaMonitor;
import git.soulbgm.pojo.dto.KcgInfo;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.*;
import java.util.concurrent.ExecutionException;

/**
 * kafka监控控制器
 *
 * @author SoulBGM
 * @date 2024-10-09
 */
@RestController
@RequestMapping("kafka")
public class KafkaMonitorController extends BaseController {

    private final AdminClient client;
    private final KafkaConsumer<?, ?> kafkaConsumer;
    private static final Map<String, String> CONSUMER_GROUP_STATE_MAP;

    static {
        CONSUMER_GROUP_STATE_MAP = new HashMap<>(6);
        CONSUMER_GROUP_STATE_MAP.put("Stable", "稳定");
        CONSUMER_GROUP_STATE_MAP.put("PreparingRebalance", "准备再平衡");
        CONSUMER_GROUP_STATE_MAP.put("CompletingRebalance", "完成再平衡");
        CONSUMER_GROUP_STATE_MAP.put("Dead", "死亡");
        CONSUMER_GROUP_STATE_MAP.put("Empty", "空");
        CONSUMER_GROUP_STATE_MAP.put("Unknown", "未知");
    }

    public KafkaMonitorController(AdminClient client, KafkaConsumer<?, ?> kafkaConsumer) {
        this.client = client;
        this.kafkaConsumer = kafkaConsumer;
    }

    @GetMapping("consumerGroups")
    public String output(String groupId) {
        try {
            List<String> groupIds;
            if (StrUtil.isBlank(groupId)) {
                groupIds = KafkaMonitor.getGroupIds(client);
            } else {
                groupIds = Collections.singletonList(groupId);
            }
            Map<String, List<KcgInfo>> map = new LinkedHashMap<>();
            for (String g : groupIds) {
                map.put(g, KafkaMonitor.getConsumerGroupInfo(client, kafkaConsumer, g));
            }
            StringBuilder sb = new StringBuilder();
            map.forEach((k, v) -> {
                if (!v.isEmpty()) {
                    KcgInfo info = v.get(0);
                    sb.append("消费组 ").append(info.getGroupId()).append(" 状态: ").append(info.getState()).append("(").append(CONSUMER_GROUP_STATE_MAP.get(info.getState())).append(")").append("\n\n");
                    sb.append("TOPIC\t\t\tPARTITION\tCURRENT-OFFSET\tLOG-END-OFFSET\tLAG\t\tHOST\n");
                    v.forEach(kcg -> sb.append(kcg.getTopic()).append("\t\t")
                            .append(kcg.getPartition()).append("\t")
                            .append(kcg.getCurrentOffset()).append("\t")
                            .append(kcg.getLatestOffset()).append("\t")
                            .append(kcg.getLag()).append("\t\t")
                            .append(kcg.getHost()).append("\n"));
                }
                sb.append("\n");
            });
            return sb.toString();
            /*
            消费组 soulStatisticsGroup 状态

            Stable(稳定)
            PreparingRebalance(准备再平衡)
            CompletingRebalance(完成再平衡)
            Dead(死亡)
            Empty(空)
            Unknown(未知)

            TOPIC                   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST
            clear_alarm             0          0               0               0               -               -
            data_segment_summary    0          363             363             0               -               -
            packet_receive_sequence 0          32              32              0               -               -
            produce_alarm           0          0               0               0               -               -
             */
        } catch (ExecutionException | InterruptedException e) {
            e.printStackTrace();
        }
        return "";
    }

}
